package com.piece.core.elasticsearch.support;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.mapping.Property;
import co.elastic.clients.elasticsearch._types.query_dsl.*;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.search.HighlightField;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.SourceConfig;
import co.elastic.clients.elasticsearch.indices.*;
import co.elastic.clients.elasticsearch.indices.ExistsRequest;
import co.elastic.clients.json.JsonData;
import com.piece.core.elasticsearch.annotations.Document;
import com.piece.core.framework.enums.sql.SqlOperator;
import com.piece.core.framework.enums.sql.SqlRelation;
import com.piece.core.framework.support.bootstrap.DataTable;
import com.piece.core.framework.support.bootstrap.Order;
import com.piece.core.framework.support.bootstrap.QueryTable;
import com.piece.core.framework.support.convert.Convert;
import com.piece.core.framework.util.basic.SpringUtil;
import com.piece.core.framework.util.collection.CollectionUtil;
import com.piece.core.framework.util.collection.MapUtil;
import com.piece.core.framework.util.object.BeanUtil;
import com.piece.core.framework.util.object.ObjectUtil;
import com.piece.core.framework.util.string.StringUtil;
import com.piece.core.framework.util.time.DateUtil;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import java.lang.reflect.Field;
import java.util.*;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Setter
public class EsSearchUtil {

    private static final String HIGHLIGHTER_PRE_TAGS = "<mark style=\"color:black;background-color:yellow;\">";
    private static final String HIGHLIGHTER_POST_TAGS = "</mark>";
    private static final ReentrantLock lock = new ReentrantLock();

    private ESClientPool clientPool = null;
    private static EsSearchUtil instance;

    public static EsSearchUtil getInstance() {
        if (null == instance) {
            lock.lock();
            if (null == instance) {
                instance = new EsSearchUtil();
                try {
                    instance.setClientPool(SpringUtil.getBean(ESClientPool.class));
                } catch (BeansException e) {
                }
            }
            lock.unlock();
        }
        return instance;
    }

    public static EsSearchUtil getInstance(ESClientPool clientPool) {
        if (null == instance) {
            lock.lock();
            if (null == instance) {
                instance = new EsSearchUtil();
                try {
                    instance.setClientPool(clientPool);
                } catch (BeansException e) {
                }
            }
            lock.unlock();
        }
        return instance;
    }

    private CreateIndexRequest indexRequest(String index, String id, Map<String, Property> properties) {
        return CreateIndexRequest.of(r -> r.index(index).mappings(m -> m.properties(properties)));
    }

    private CreateRequest createRequest(String index, String id, Object document) {
        return CreateRequest.of(r -> r.index(index).id(id).document(document));
    }

    private UpdateRequest updateRequest(String index, String id, Object document) {
        return UpdateRequest.of(r -> r.index(index).id(id).doc(document));
    }

    private DeleteRequest deleteRequest(String index, String id) {
        return DeleteRequest.of(r -> r.index(index).id(id));
    }

    private DeleteIndexRequest deleteIndexRequest(String index) {
        return DeleteIndexRequest.of(r -> r.index(index));
    }

    private PutMappingRequest mappingRequest(String index, Map<String, Property> properties) {
        return PutMappingRequest.of(r -> r.index(index)
                .properties(properties));
    }

    private boolean checkIndex(ElasticsearchClient client, String index) {
        ExistsRequest existsRequest = ExistsRequest.of(r -> r.index(index));
        try {
            return client.indices().exists(existsRequest).value();
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * 插入索引数据
     */
    public void insert(Object document) throws Exception {
        if (null != clientPool) {
            ElasticsearchClient client = clientPool.getClient();
            String index = convertIndex(document.getClass());

            try {
                client.indices().create(indexRequest(index, String.valueOf(BeanUtil.getKeyValue(document)), convertProperty(document.getClass())));
                client.create(createRequest(index, String.valueOf(BeanUtil.getKeyValue(document)), document));
            } catch (Exception e) {
                log.error(e.getMessage());
            } finally {
                clientPool.returnClient(client);
            }
        }
    }

    /**
     * 插入索引数据
     */
    public void insertBatch(List list, Class clazz) throws Exception {
        if (null != clientPool) {
            String index = convertIndex(clazz);
            ElasticsearchClient client = clientPool.getClient();

            try {
                List<BulkOperation> bulkOperations = new ArrayList<>();
                for (Object document: list) {
                    bulkOperations.add(BulkOperation.of(o -> o
                            .index(i -> i.id(String.valueOf(BeanUtil.getKeyValue(document))).document(document))));
                }
                client.bulk(r -> r.index(index).operations(bulkOperations));
            } catch (Exception e) {
                log.error(e.getMessage());
            } finally {
                clientPool.returnClient(client);
            }
        }
    }

    /**
     * 更新索引数据
     */
    public void update(Object document) throws Exception {
        if (null != clientPool) {
            String index = convertIndex(document.getClass());
            ElasticsearchClient client = clientPool.getClient();

            try {
                client.update(updateRequest(index, String.valueOf(BeanUtil.getKeyValue(document)), document),
                        document.getClass());
            } catch (Exception e) {
                log.error(e.getMessage());
            } finally {
                clientPool.returnClient(client);
            }
        }
    }

    /**
     * 更新索引数据
     */
    public void updateBatch(List list, Class clazz) throws Exception {
        if (null != clientPool) {
            String index = convertIndex(clazz);
            ElasticsearchClient client = clientPool.getClient();

            try {
                List<BulkOperation> bulkOperations = new ArrayList<>();
                list.forEach(document -> {
                    bulkOperations.add(BulkOperation.of(o -> o
                            .update(u -> u
                                    .id(String.valueOf(BeanUtil.getKeyValue(document)))
                                    .action(a -> a.doc(document)))));
                });

                client.bulk(b -> b.index(index).operations(bulkOperations));
            } catch (Exception e) {
                log.error(e.getMessage());
            } finally {
                clientPool.returnClient(client);
            }
        }
    }

    /**
     * 删除索引数据
     */
    public void delete(String id, Class clazz) throws Exception {
        if (null != clientPool) {
            String index = convertIndex(clazz);
            ElasticsearchClient client = clientPool.getClient();

            try {
                client.delete(deleteRequest(index, id));
            } catch (Exception e) {
                log.error(e.getMessage());
            } finally {
                clientPool.returnClient(client);
            }
        }
    }

    /**
     * 删除索引数据
     */
    public void delete(Class clazz) throws Exception {
        if (null != clientPool) {
            String index = convertIndex(clazz);
            ElasticsearchClient client = clientPool.getClient();

            try {
                client.indices().delete(deleteIndexRequest(index));
            } catch (Exception e) {
                log.error("Elasticsearch 删除索引[{}]失败: {}", index, e.getMessage());
            } finally {
                clientPool.returnClient(client);
            }
        }
    }

    /**
     * 删除索引数据
     */
    public void delete(Object document) throws Exception {
        if (null != clientPool) {
            ElasticsearchClient client = clientPool.getClient();
            String index = convertIndex(document.getClass());
            String id = String.valueOf(BeanUtil.getKeyValue(document));

            try {
                if (StringUtil.isNotEmpty(id)) {
                    client.delete(deleteRequest(index, id));
                } else {
                    SearchResponse searchResponse = search(index, searchRequestBuilderWithEntity(document), document.getClass());
                    searchResponse.hits().hits().forEach(search -> {
                        try {
                            client.delete(deleteRequest(index, id));
                        } catch (Exception e) {
                            log.error("Elasticsearch 删除索引[{}]数据[{}]失败: {}", index, id, e.getMessage());
                        }
                    });
                }
            } catch (Exception e) {
                log.error(e.getMessage());
            } finally {
                clientPool.returnClient(client);
            }
        }
    }

    private Query query() {
        return Query.of(q -> q.matchAll(m -> m));
    }

    private SearchRequest.Builder searchRequestBuilderWithEntity(Object document) {
        SearchRequest.Builder builder = new SearchRequest.Builder().index(convertIndex(document.getClass()));
        if (ObjectUtil.isEmpty(document)) {
            return builder;
        }

        Query.Builder queryBuilder = new Query.Builder();
        Arrays.stream(document.getClass().getDeclaredFields()).forEach(field -> {
            String name = field.getName();
            Object value = BeanUtil.getFieldValue(document, name);
            if (null != value) {
                queryBuilder.term(TermQuery.of(t -> t.field(name).value(FieldValue.of(JsonData.of(value)))));
            }
        });

        return builder.query(queryBuilder.build());
    }

    private SearchRequest.Builder searchRequestBuilderWithParam(Map<String, Object> searchParams, Class clazz) {
        SearchRequest.Builder builder = new SearchRequest.Builder().index(convertIndex(clazz));
        if (MapUtil.isEmpty(searchParams)) {
            return builder;
        }

        BoolQuery.Builder boolBuilder = new BoolQuery.Builder();
        List<Query> mustQueryList = new ArrayList<>();
        List<Query> shouldQueryList = new ArrayList<>();
        List<Query> mustNotQueryList = new ArrayList<>();
        boolean isOr = false;
        Query query = null;
        for (Map.Entry<String, Object> entry : searchParams.entrySet()) {
            // 过滤掉空值
            String name = entry.getKey();
            Object value = entry.getValue();

            if (!name.startsWith("Q_")) {
                mustQueryList.add(Query.of(q -> q.term(TermQuery.of(t -> t.field(name)
                        .value(FieldValue.of(JsonData.of(value)))))));
            } else {
                String[] names = StringUtils.split(name, "_");
                String filedName = names[1];
                SqlOperator operator = SqlOperator.valueOf(names[3]);
                if (SqlOperator.NULL != operator && SqlOperator.NNULL != operator
                        && SqlOperator.EMP != operator && SqlOperator.NEMP != operator
                        && null == value) {
                    continue;
                }

                // 拆分operator、relation、filedAttribute
                boolean hasRelation = (name.split("_").length == 5);
                SqlRelation relation = SqlRelation.AND;
                String symbol = null;
                if (hasRelation) {
                    symbol = names[4];
                    relation = SqlRelation.valueOf(symbol.toUpperCase());
                }

                //symbol = names[2];
                String analyzer = null;

                switch (operator) {
                    case NNULL:
                    case NEMP:
                        query = Query.of(b -> b.wildcard(q -> q.field(filedName).value("*")));
                        if (relation.equals(SqlRelation.AND)) {
                            mustQueryList.add(query);
                        } else {
                            isOr = true;
                            shouldQueryList.add(query);
                        }
                        break;
                    case EMP:
                        query = Query.of(b -> b.term(q -> q.field(filedName).value("")));
                        if (relation.equals(SqlRelation.AND)) {
                            mustQueryList.add(query);
                        } else {
                            isOr = true;
                            shouldQueryList.add(query);
                        }
                    case NULL:
                        query = Query.of(b -> b.exists(q -> q.field(filedName)));
                        if (relation.equals(SqlRelation.AND)) {
                            mustQueryList.add(query);
                        } else {
                            isOr = true;
                            shouldQueryList.add(query);
                        }
                        break;
                    case EQ:
                        query = Query.of(b -> b.term(q -> q.field(filedName)
                                .value(FieldValue.of(JsonData.of(value)))));
                        if (relation.equals(SqlRelation.AND)) {
                            mustQueryList.add(query);
                        } else {
                            isOr = true;
                            shouldQueryList.add(query);
                        }
                        break;
                    case NEQ:
                        query = Query.of(b -> b.term(q -> q.field(filedName)
                                .value(FieldValue.of(JsonData.of(value)))));
                        mustNotQueryList.add(query);
                        break;
                    case IN:
                        query = Query.of(b -> b.terms(q -> q.terms(t -> t.value(convertFieldValue(value)))));
                        if (relation.equals(SqlRelation.AND)) {
                            mustQueryList.add(query);
                        } else {
                            isOr = true;
                            shouldQueryList.add(query);
                        }
                        break;
                    default:
                        try {
                            Field field = BeanUtil.getFieldByFieldName(clazz, filedName);
                            com.piece.core.elasticsearch.annotations.Field source = field.getAnnotation(com.piece.core.elasticsearch.annotations.Field.class);
                            if (null != source && source.searchAnalyzer()) {
                                analyzer = source.analyzer().getType();
                            }
                        } catch (Exception e) {}

                        MatchQuery.Builder matchQueryBuilder = new MatchQuery.Builder();
                        if (StringUtil.isNotEmpty(analyzer)) {
                            matchQueryBuilder.analyzer(analyzer);
                        }

                        query = Query.of(b -> b.match(q -> matchQueryBuilder.field(filedName).query(FieldValue.of(JsonData.of(value)))));
                        if (relation.equals(SqlRelation.AND)) {
                            mustQueryList.add(query);
                        } else {
                            isOr = true;
                            shouldQueryList.add(query);
                        }
                }
            }
        }

        if (isOr) {
            boolBuilder.should(shouldQueryList);
        }

        if (CollectionUtil.isNotEmpty(mustNotQueryList)) {
            boolBuilder.mustNot(mustNotQueryList);
        }

        return builder.query(new Query.Builder()
                .bool(boolBuilder.must(mustQueryList).build()).build());
    }

    private List<FieldValue> convertFieldValue(Object value) {
        List<FieldValue> fieldValues = new ArrayList<>();
        Arrays.stream(Convert.toStrArray(Convert.toStr(value))).forEach(value_ -> {
            fieldValues.add(FieldValue.of(JsonData.of(value_)));
        });
        return fieldValues;
    }

    private SearchRequest searchRequest(String index) {
        return SearchRequest.of(r -> r.index(index));
    }

    private SearchRequest searchRequest(String index, String queryStr) {
        return SearchRequest.of(r -> r.index(index).q(queryStr));
    }

    private SearchRequest searchRequest(String index, SearchRequest.Builder builder) {
        return SearchRequest.of(r -> builder);
    }

    private SearchResponse search(String index, Class clazz) throws Exception {
        SearchRequest searchRequest = searchRequest(index);
        SearchResponse searchResponse = null;
        ElasticsearchClient client = clientPool.getClient();

        try {
            searchResponse = client.search(searchRequest, clazz);
        } catch (Exception e) {
            log.error("Elasticsearch 搜索索引{}数据失败", index);
        } finally {
            clientPool.returnClient(client);
        }

        return searchResponse;
    }

    private SearchResponse search(String index, SearchRequest.Builder builder, Class clazz) throws Exception {
        SearchRequest searchRequest = searchRequest(index, builder);
        SearchResponse searchResponse = null;
        ElasticsearchClient client = clientPool.getClient();

        try {
            searchResponse = client.search(searchRequest, clazz);
        } catch (Exception e) {
            log.error("Elasticsearch 搜索索引{}数据失败", index);
        } finally {
            clientPool.returnClient(client);
        }

        return searchResponse;
    }

    /**
     * 分页查询
     *
     * @param queryTable  分页条件
     * @param trackTotalHits 总数是否显示所有数据，默认只显示10000条
     * @param isHighlighter 是否高亮标注
     */
    public <T> DataTable searchPage(QueryTable queryTable, boolean trackTotalHits, boolean isHighlighter) {
        Class clazz = queryTable.getEntity();
        DataTable dataTable = new DataTable(queryTable);
        dataTable.setData(Collections.emptyList());
        dataTable.setRecordsTotal(0);

        if (null != clientPool) {
            try {
                SearchRequest.Builder builder = searchRequestBuilderWithParam(queryTable.getParams(), clazz);
                if (queryTable.getLength() != 0) {
                    builder.from(queryTable.getStart()).size(queryTable.getLength());
                    if (trackTotalHits) {
                        builder.trackTotalHits(h -> h);
                    }
                }

                if (null != queryTable.getOrder()) {
                    queryTable.getOrder().forEach(order -> {
                        addSort(builder, order);
                    });
                }

                setIsHighlight(builder, queryTable.getParams(), isHighlighter);
                String index = convertIndex(clazz);
                SearchResponse response = search(index, builder, clazz);
                long totalCnt = response.hits().total().value();
                List<T> list = searchList(response);
                dataTable.setData(list);
                dataTable.setRecordsTotal(Convert.toInt(totalCnt, 0));
            } catch (Exception e) {
                log.error("Elasticsearch 分页查询数据失败: {}", e.getMessage());
            }
        }

        return dataTable;
    }

    public List searchList(Map<String, Object> param, List<Order> orders, boolean isHighlighter, Class clazz) {
        if (null != clientPool) {
            try {
                String index = convertIndex(clazz);
                SearchRequest.Builder builder = searchRequestBuilderWithParam(param, clazz);
                if (null != orders) {
                    orders.forEach(order -> {
                        addSort(builder, order);
                    });
                }

                setIsHighlight(builder, param, isHighlighter);
                SearchResponse response = search(index, builder, clazz);
                return searchList(response);
            } catch (Exception e) {
                log.error("Elasticsearch 查询数据失败: {}", e.getMessage());
            }
        }

        return new ArrayList();
    }

    private <T> List searchList(SearchResponse response) {
        List list = new ArrayList<>();

        List<Hit<T>> hits = response.hits().hits();
        for (Hit<T> hit : hits) {
            T document = hit.source();
            Map<String, List<String>> highlight = hit.highlight();
            if (null != highlight) {
                populateHighLightedFields(document, highlight);
            }
            list.add(document);
        }

        return list;
    }

    /**
     * 组装高亮字符
     *
     * @param result 目标对象
     * @param highlightFields 高亮配置
     */
    private <T> void populateHighLightedFields(T result, Map<String, List<String>> highlightFields) {
        highlightFields.keySet().stream().forEach(name -> {
            if (!name.endsWith(".keyword")) {
                try {
                    BeanUtil.setFieldValue(result, name, highlightFields.get(name).get(0));
                } catch (Exception e) {
                    log.error("failed to set highlighted value for field: [{}] with value: [{}]", name, highlightFields.get(name), e);
                }
            }
        });
    }

    private void addSort(SearchRequest.Builder builder, Order order) {
        if (null != order && StringUtil.isNotEmpty(order.getAttr())) {
            switch (order.getSort()) {
                case ASC:
                    builder.sort(s -> s.field(f -> f.order(SortOrder.Asc).field(order.getAttr())));
                    break;
                case DESC:
                default:
                    builder.sort(s -> s.field(f -> f.order(SortOrder.Desc).field(order.getAttr())));
            }
        }
    }

    /**
     * 设置是否需要高亮处理
     */
    private SearchRequest.Builder setIsHighlight(SearchRequest.Builder builder, Map<String, Object> params, boolean isHighlighter) {
        if (isHighlighter) {
            builder.source(SourceConfig.of(source -> source.fetch(true)));

            Map<String, HighlightField> highlightFields = new HashMap<>();
            for (Map.Entry<String, Object> entry : params.entrySet()) {
                String name = entry.getKey();
                if (!name.startsWith("Q_")) {
                    highlightFields.put(name, HighlightField.of(f -> f));
                } else {
                    highlightFields.put(StringUtils.split(name, "_")[1], HighlightField.of(f -> f));
                }
            }
            builder.highlight(h -> h.preTags(HIGHLIGHTER_PRE_TAGS)
                    .postTags(HIGHLIGHTER_POST_TAGS)
                    .requireFieldMatch(false)
                    .fields(highlightFields));
        }

        return builder;
    }

    private String convertIndex(Class clazz) {
        Document document = (Document) clazz.getAnnotation(Document.class);
        if (null == document) {
            return clazz.getSimpleName().toLowerCase();
        }

        return document.indexName();
    }

    private Map<String, Object> convertToMap(Object document) {
        Map<String, Object> result = new HashMap<>();
        Field[] fields = document.getClass().getDeclaredFields();
        Arrays.stream(fields).forEach(field -> {
            try {
                String name = null;
                field.setAccessible(true);
                Object value = field.get(document);
                com.piece.core.elasticsearch.annotations.Field source = field.getAnnotation(com.piece.core.elasticsearch.annotations.Field.class);
                if (null != source) {
                    name = StringUtil.isNotEmpty(source.name()) ? source.name() : field.getName();
                    result.put(name, value);
                }
            } catch (IllegalAccessException e) {
            }
        });

        return result;
    }

    private Map<String, Property> convertProperty(Class clazz) {
        Map<String, Property> builder = new HashMap<>();
        Field[] fields = clazz.getDeclaredFields();
        try {
            for (int i = 0; i < fields.length; i++) {
                Field field = fields[i];
                String name = null;
                com.piece.core.elasticsearch.annotations.Field source = field.getAnnotation(com.piece.core.elasticsearch.annotations.Field.class);
                if (null != source && source.searchAnalyzer()) {
                    name = StringUtil.isNotEmpty(source.name()) ? source.name() : field.getName();
                    Property property = null;
                    switch (source.type()) {
                        case KEYWORD:
                            property = Property.of(p -> p.keyword(k -> k));
                            break;
                        case TEXT:
                            property = Property.of(p -> p.text(t -> t.analyzer(source.analyzer().getType())));
                            break;
                        case DATE:
                            property = Property.of(p -> p.date(d -> d.format(DateUtil.FORMAT_YYYY_MM_DD_HH_MM_SS_E)));
                            break;
                        case INTEGER:
                            property = Property.of(p -> p.integer(i_ -> i_.nullValue(0)));
                            break;
                        case DOUBLE:
                            property = Property.of(p -> p.double_(d -> d.nullValue(0.00)));
                            break;
                        default:
                            property = Property.of(p -> p);
                            break;
                    }

                    builder.put(name, property);
                }
            }
        } catch (Exception e) {
        }

        return builder;
    }
}
